package shard

import akka.actor.{Actor, ActorLogging, OneForOneStrategy, Props, SupervisorStrategy}
import akka.cluster.Cluster
import akka.cluster.sharding.ShardRegion
import akka.persistence.{PersistentActor, Recovery, SnapshotOffer}
import shard.Calculator._

import scala.concurrent.duration._

/**
 * @Description: TODO
 * @Author: DJJ
 * @Date: 2019/12/26 20:15
 */
object Calculator {

  sealed trait Command

  case class Num(d: Double) extends Command

  case class Add(d: Double) extends Command

  case class Sub(d: Double) extends Command

  case class Mul(d: Double) extends Command

  case class Div(d: Double) extends Command

  case object ShowResult extends Command


  sealed trait Event

  case class SetResult(d: Any) extends Event

  def getResult(res: Double, cmd: Command) = cmd match {
    case Num(x) => x
    case Add(x) => res + x
    case Sub(x) => res - x
    case Mul(x) => res * x
    case Div(x) => {
      val _ = res.toInt / x.toInt //yield ArithmeticException when /0.00
      res / x
    }
    case _ => new ArithmeticException("Invalid Operation!")
  }

  case class State(result: Double) {
    //更新状态
    def updateState(evt: Event): State = evt match {
      case SetResult(n) => copy(result = n.asInstanceOf[Double])
    }
  }


  case object Disconnect extends Command //exit cluster
  //TODO
  def props = Props(new Calcultor)

}

/**
 *  持久化
 */
class Calcultor extends PersistentActor with ActorLogging {

  var cluster = Cluster(context.system)

  var state: State = State(0);


  override def persistenceId: String = self.path.parent.name + "-------" + self.path.name

  //在恢复过程中处理事件和快照消息。
  override def receiveRecover: Receive = {
    case evt: Event =>
      println("333333333333")
      state = state.updateState(evt)
    case SnapshotOffer(_, st: State) =>
      println("SnapshotOffer-------")
      state = state.copy(result = st.result)
  }

  override def receiveCommand: Receive = {
    case Num(n) => persist(SetResult(getResult(state.result, Num(n))))(evt => state = state.updateState(evt))
    case Add(n) => persist(SetResult(getResult(state.result, Add(n))))(evt => state = state.updateState(evt))
    case Sub(n) => persist(SetResult(getResult(state.result, Sub(n))))(evt => state = state.updateState(evt))
    case Mul(n) => persist(SetResult(getResult(state.result, Mul(n))))(evt => state = state.updateState(evt))
    case Div(n) => persist(SetResult(getResult(state.result, Div(n))))(evt => state = state.updateState(evt))
    case ShowResult =>
      println("111111111111111")
      log.info(s"Result on ${cluster.selfAddress.hostPort} is: ${state.result}")
    case Disconnect =>
      log.info(s"${cluster.selfAddress} is leaving cluster!!!")
      cluster.leave(cluster.selfAddress)

  }

  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    log.info(s"Restarting calculator: ${reason.getMessage}")
    super.preRestart(reason, message)
  }

  override def recovery: Recovery = super.recovery


}

class CalcSupervisor extends Actor {
  def decider: PartialFunction[Throwable, SupervisorStrategy.Directive] = {
    case _: ArithmeticException => SupervisorStrategy.Resume
  }

  override def supervisorStrategy: SupervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 5, withinTimeRange = 5 seconds) {
      decider.orElse(SupervisorStrategy.defaultDecider)
    }

  val calcActor = context.actorOf(Calculator.props, "calculator")

  override def receive: Receive = {
    case msg@_ => calcActor.forward(msg)
  }

}

object CalculatorShard {

  import Calculator._

  case class CalcCommands(eid: String, msg: Command) //user should use it to talk to shardregion
  val shardName = "calcShard"
  val getEntityId: ShardRegion.ExtractEntityId = {
    case CalcCommands(id, msg) =>
      println("ExtractEntityId-------:"+id)
      (id, msg)
  }
  val getShardId: ShardRegion.ExtractShardId = {
    case CalcCommands(id, _) =>
      println("ExtractShardId-------:"+id)
      id.head.toString
  }

  def entityProps = Props(new CalcSupervisor)
}
